package com.lims.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.digest.MD5;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import com.lims.base.entity.TestUser;
import com.lims.base.mapper.TestUserMapper;
import com.lims.common.RedisKeyConts;
import com.lims.exception.BaseException;
import com.lims.kafka.TopicConts;
import com.lims.service.ITestUserService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class TestUserServiceImpl extends ServiceImpl<TestUserMapper, TestUser> implements ITestUserService {

    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    @Cacheable(value = RedisKeyConts.LIMS_TEST + "testUser:getOneWithCache#120", unless = "#result == null")
    public TestUser getOneWithCache(TestUser deviceEntity) {
        QueryWrapper<TestUser> queryWrapper = Wrappers.query(deviceEntity);
        return getOne(queryWrapper);
    }

    @Override
    public void dealMessageList(List<String> messageList) {
        List<TestUser> entityList = new ArrayList<>();
        for(String message : messageList){
            TestUser entity = JSONUtil.toBean(message, TestUser.class);
            //todo 做其它业务处理
            entityList.add(entity);
        }
        //批量入库
        this.saveBatch(entityList);
    }

    @Override
    public void receive(String req) {
        //重复校验
        String md5 = new MD5().digestHex16(req);
        String value = stringRedisTemplate.opsForValue().get(md5);
        if (StrUtil.isNotBlank(value) && "1".equals(value)) {
            log.info("TestUserServiceImpl.receive重复推送");
            throw new BaseException("重复推送");
        }
        //缓存，防重（使用redis缓存， key = md5Hex16， value = 1）,两分钟内
        stringRedisTemplate.opsForValue().set(md5, "1", 2, TimeUnit.MINUTES);
        /**
         * 业务处理
         */
        TestUser entity = JSONUtil.toBean(req, TestUser.class);
        //todo

        //处理后的数据转成json字符串推到kafka
        kafkaTemplate.send(TopicConts.TEST_TOPIC, JSON.toJSONString(entity));
    }
}
